package drds.global_transaction.api.rpc;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.alibaba.fastjson.JSON;

import drds.common.Constants;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.IMergeMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessageFuture;
import drds.global_transaction.api.rpc.config.ClientConfig;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class AbstractRpcChannelDuplexHandler extends ChannelDuplexHandler
{
	private static final long not_writeable_check_mills = 10L;
	private static final int timeout_check_internal = 3000;
	/**
	 * The Timer executor.
	 */
	protected final ScheduledExecutorService scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("timeoutChecker", 1, true));
	/**
	 * The Message executor.
	 */
	protected final ThreadPoolExecutor threadPoolExecutor;
	private final Object lock = new Object();
	@Setter
	@Getter
	public ConcurrentHashMap<Long, RpcMessageFuture> rpcMessageIdToRpcMessageFutureMap = new ConcurrentHashMap<Long, RpcMessageFuture>();
	@Setter
	@Getter
	public ConcurrentHashMap<String, BlockingQueue<RpcMessage>> addressToRpcMessageQueueMap = new ConcurrentHashMap<String, BlockingQueue<RpcMessage>>();
	@Setter
	@Getter
	public Object mergedWarpMessageLock = new Object();
	@Setter
	@Getter
	public Map<Long, IMergeMessage> messageIdToMergedWarpMessageMap = new ConcurrentHashMap<Long, IMergeMessage>();
	@Setter
	@Getter
	public volatile long nowDateTimeLong = 0;
	@Setter
	@Getter
	public boolean isSending = false;
	@Setter
	@Getter
	public ChannelHandler[] channelHandlers;
	/**
	 * 用于测试。发现线程池满时可以打开这个变量，把堆栈打出来分享
	 */
	@Setter
	@Getter
	boolean allowDumpStack = false;
	@Setter
	@Getter
	private String group = "DEFAULT";

	public AbstractRpcChannelDuplexHandler(ThreadPoolExecutor threadPoolExecutor)
	{
		this.threadPoolExecutor = threadPoolExecutor;
	}

	public void init()
	{
		scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				List<RpcMessageFuture> timeoutRpcMessageFutureList = new ArrayList<RpcMessageFuture>(rpcMessageIdToRpcMessageFutureMap.size());

				for (RpcMessageFuture rpcMessageFuture : rpcMessageIdToRpcMessageFutureMap.values())
				{
					if (rpcMessageFuture.isTimeout())
					{
						timeoutRpcMessageFutureList.add(rpcMessageFuture);
					}
				}

				for (RpcMessageFuture rpcMessageFuture : timeoutRpcMessageFutureList)
				{
					rpcMessageIdToRpcMessageFutureMap.remove(rpcMessageFuture.getRequestMessage().getId());
					rpcMessageFuture.setResultMessage(null);
					if (log.isDebugEnabled())
					{
						log.debug("timeout clear future : " + rpcMessageFuture.getRequestMessage().getMessageCodec());
					}
				}
				nowDateTimeLong = System.currentTimeMillis();
			}
		}, timeout_check_internal, timeout_check_internal, TimeUnit.MILLISECONDS);
	}

	public void destroy()
	{
		scheduledThreadPoolExecutor.shutdown();
	}

	@Override
	public void channelWritabilityChanged(ChannelHandlerContext channelHandlerContext)
	{
		synchronized (lock)
		{
			if (channelHandlerContext.channel().isWritable())
			{
				lock.notifyAll();
			}
		}

		channelHandlerContext.fireChannelWritabilityChanged();
	}

	public Object sendAsynchronousRequestWithResponse(String address, Channel channel, Object message) throws TimeoutException
	{
		return sendAsynchronousRequestWithResponse(address, channel, message, ClientConfig.getRpcRequestTimeout());
	}

	protected Object sendAsynchronousRequestWithResponse(String address, Channel channel, Object message, long timeout) throws TimeoutException
	{
		if (timeout <= 0)
		{
			throw new FrameworkException("timeout should more than 0ms");
		}
		return asynchronousSendRequest(address, channel, message, timeout);
	}

	protected Object asynchronousSendRequestWithoutResponse(String address, Channel channel, Object message) throws TimeoutException
	{
		return asynchronousSendRequest(address, channel, message, 0);
	}

	private Object asynchronousSendRequest(String address, Channel channel, Object message, long timeout) throws TimeoutException
	{
		if (channel == null)
		{
			log.warn("sendAsynchronousRequestWithResponse nothing, caused by null channel.");
			return null;
		}
		final RpcMessage rpcMessage = new RpcMessage();
		rpcMessage.setId(RpcMessage.getNextMessageId());
		rpcMessage.setAsync(false);
		rpcMessage.setHeartbeat(false);
		rpcMessage.setRequest(true);
		rpcMessage.setMessageCodec(message);

		final RpcMessageFuture rpcMessageFuture = new RpcMessageFuture();
		rpcMessageFuture.setRequestMessage(rpcMessage);
		rpcMessageFuture.setTimeout(timeout);
		rpcMessageIdToRpcMessageFutureMap.put(rpcMessage.getId(), rpcMessageFuture);

		if (address != null)
		{
			BlockingQueue<RpcMessage> rpcMessageQueue = addressToRpcMessageQueueMap.get(address);
			if (rpcMessageQueue == null)
			{
				addressToRpcMessageQueueMap.putIfAbsent(address, new LinkedBlockingQueue<RpcMessage>());
				rpcMessageQueue = addressToRpcMessageQueueMap.get(address);
			}
			rpcMessageQueue.offer(rpcMessage);
			// not sending message now,if has data ,then deal with it,otherwise not deal
			// with it
			if (!isSending)
			{
				// notice deal with these message list
				synchronized (mergedWarpMessageLock)
				{
					mergedWarpMessageLock.notifyAll();
				}
			}
		} else
		{

			channelWriteableCheck(channel, message);
			ChannelFuture channelFuture = channel.writeAndFlush(rpcMessage);
			channelFuture.addListener(new ChannelFutureListener()
			{
				@Override
				public void operationComplete(ChannelFuture channelFuture)
				{
					if (!channelFuture.isSuccess())
					{
						RpcMessageFuture rpcMessageFuture = rpcMessageIdToRpcMessageFutureMap.remove(rpcMessage.getId());
						if (rpcMessageFuture != null)
						{
							rpcMessageFuture.setResultMessage(channelFuture.cause());
						}
						returnChannel(channelFuture.channel());
					}
				}
			});
		}
		if (timeout > 0)
		{
			try
			{
				Object object = rpcMessageFuture.get(timeout, TimeUnit.MILLISECONDS);
				if (Constants.developMode)
				{
					log.debug("服务器返回信息:" + object);
					if (object != null)
					{
						log.debug(JSON.toJSONString(object));
					}
				}
				return object;
			} catch (Exception exx)
			{
				log.error("wait response error:" + exx.getMessage() + ",ip:" + address + ",request:" + message);
				if (exx instanceof TimeoutException)
				{
					throw (TimeoutException) exx;
				} else
				{
					throw new RuntimeException(exx);
				}
			}
		} else
		{
			return null;
		}
	}

	public void sendRequest(Channel channel, Object message)
	{
		RpcMessage rpcMessage = new RpcMessage();
		rpcMessage.setAsync(true);
		rpcMessage.setHeartbeat(message instanceof HeartbeatMessage);
		rpcMessage.setRequest(true);
		rpcMessage.setMessageCodec(message);
		rpcMessage.setId(RpcMessage.getNextMessageId());
		if (message instanceof IMergeMessage)
		{
			messageIdToMergedWarpMessageMap.put(rpcMessage.getId(), (IMergeMessage) message);
		}
		channelWriteableCheck(channel, message);
		channel.writeAndFlush(rpcMessage);
	}

	protected void sendResponse(long msgId, Channel channel, Object message)
	{
		RpcMessage rpcMessage = new RpcMessage();
		rpcMessage.setAsync(true);
		rpcMessage.setHeartbeat(message instanceof HeartbeatMessage);
		rpcMessage.setRequest(false);
		rpcMessage.setMessageCodec(message);
		rpcMessage.setId(msgId);
		channelWriteableCheck(channel, message);
		channel.writeAndFlush(rpcMessage);
	}

	private void channelWriteableCheck(Channel channel, Object message)
	{
		int tryTimes = 0;
		synchronized (lock)
		{
			while (!channel.isWritable())
			{
				try
				{
					tryTimes++;
					if (tryTimes > ClientConfig.getMaxNotWriteableRetry())
					{
						returnChannel(channel);
						throw new FrameworkException("msg:" + ((message == null) ? "null" : message.toString()), FrameworkErrorCode.ChannelIsNotWritable);
					}
					lock.wait(not_writeable_check_mills);
				} catch (InterruptedException exx)
				{
					log.error(exx.getMessage());
				}
			}
		}
	}

	@Override
	public void channelRead(final ChannelHandlerContext channelHandlerContext, Object message) throws Exception
	{
		if (message instanceof RpcMessage)
		{
			final RpcMessage rpcMessage = (RpcMessage) message;
			if (rpcMessage.isRequest())
			{
				if (log.isDebugEnabled())
				{
					log.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getMessageCodec()));
				}
				try
				{
					AbstractRpcChannelDuplexHandler.this.threadPoolExecutor.execute(new Runnable()
					{
						@Override
						public void run()
						{
							try
							{
								dispatch(rpcMessage.getId(), channelHandlerContext, rpcMessage.getMessageCodec());
							} catch (Throwable th)
							{
								log.error(FrameworkErrorCode.NetDispatch.errCode, th.getMessage(), th);
							}
						}
					});
				} catch (RejectedExecutionException e)
				{
					log.error(FrameworkErrorCode.ThreadPoolFull.errCode, "thread pool is full, current max pool size is " + threadPoolExecutor.getActiveCount());
					if (allowDumpStack)
					{
						String name = ManagementFactory.getRuntimeMXBean().getName();
						String pid = name.split("@")[0];
						int idx = new Random().nextInt(100);
						try
						{
							Runtime.getRuntime().exec("jstack " + pid + " >d:/" + idx + ".log");
						} catch (IOException exx)
						{
							log.error(exx.getMessage());
						}
						allowDumpStack = false;
					}
				}
			} else
			{
				RpcMessageFuture rpcMessageFuture = rpcMessageIdToRpcMessageFutureMap.remove(rpcMessage.getId());
				if (log.isDebugEnabled())
				{
					log.debug(String.format("%s msgId:%s, future :%s, body:%s", this, rpcMessage.getId(), rpcMessageFuture, rpcMessage.getMessageCodec()));
				}
				if (rpcMessageFuture != null)
				{
					rpcMessageFuture.setResultMessage(rpcMessage.getMessageCodec());
				} else
				{
					try
					{
						AbstractRpcChannelDuplexHandler.this.threadPoolExecutor.execute(new Runnable()
						{
							@Override
							public void run()
							{
								try
								{
									dispatch(rpcMessage.getId(), channelHandlerContext, rpcMessage.getMessageCodec());
								} catch (Throwable th)
								{
									log.error(FrameworkErrorCode.NetDispatch.errCode, th.getMessage(), th);
								}
							}
						});
					} catch (RejectedExecutionException e)
					{
						log.error(FrameworkErrorCode.ThreadPoolFull.errCode, "thread pool is full, current max pool size is " + threadPoolExecutor.getActiveCount());
					}
				}
			}
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		log.error(FrameworkErrorCode.ExceptionCaught.errCode, channelHandlerContext.channel() + " connect exception. " + cause.getMessage(), cause);
		try
		{
			returnChannel(channelHandlerContext.channel());
		} catch (Exception e)
		{
			log.error("", "close channel" + channelHandlerContext.channel() + " fail.", e);
		}
	}

	public abstract void dispatch(long messageId, ChannelHandlerContext channelHandlerContext, Object message);

	@Override
	public void close(ChannelHandlerContext channelHandlerContext, ChannelPromise channelPromise) throws Exception
	{
		super.close(channelHandlerContext, channelPromise);
	}

	protected void addChannelHandlersToChannelPipelineLast(Channel channel, ChannelHandler... channelHandlers)
	{
		if (null != channel && null != channelHandlers)
		{
			channel.pipeline().addLast(channelHandlers);
		}
	}

	public void returnChannel(Channel channel)
	{
		returnChannel(getRemoteAddress(channel), channel);
	}

	public abstract void returnChannel(String serverAddress, Channel channel);

	protected String getRemoteAddress(ChannelHandlerContext channelHandlerContext)
	{
		return getRemoteAddress(channelHandlerContext.channel());
	}

	protected String getRemoteAddress(Channel channel)
	{
		SocketAddress socketAddress = channel.remoteAddress();
		String address = socketAddress.toString();
		if (socketAddress.toString().indexOf(ClientConfig.getSocketAddressStartChar()) == 0)
		{
			address = socketAddress.toString().substring(ClientConfig.getSocketAddressStartChar().length());
		}
		return address;
	}
}
